package com.xiaohong.mq.listener;

import com.alibaba.fastjson.JSON;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.GetResponse;
import com.xiaohong.mq.pojo.ReceiveMsg;
import com.xiaohong.mq.pojo.SendMsg;
import org.springframework.web.bind.annotation.*;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeoutException;

/**
 * @author : xiaohong
 * @date : 2021/4/14 16:04
 * 手动拉取消息
 */

@RestController
@CrossOrigin
@RequestMapping("/pull")
public class PullMessage {

    // 所有的中间件技术都是基于TCP/IP协议基础上构建的协议规范，只不过rabbitmq遵循的是amqp

    @GetMapping("/nack/{cid}")
    public List<ReceiveMsg> pullNack(@PathVariable String cid) {
        //创建连接工程
        ConnectionFactory connectionFactory  = new ConnectionFactory();

        connectionFactory.setHost("119.29.67.157");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("xiaohong");
        connectionFactory.setPassword("xiaohong");
        connectionFactory.setVirtualHost("my_vhost");


        Connection connection = null;
        Channel channel = null;
        //存放消息实体List
        List<ReceiveMsg> list = new ArrayList<>();
        try {
            //创建连接
            connection = connectionFactory.newConnection();
            // 通过连接获取通道Channel
            channel = connection.createChannel();
            // 通过通道创建交换机，声明队列，交换关系，路由key，发送消息和接受消息
            // 队列名称
            String queueName = "chat";
            // 手动拉取一条消息
            // 参数2 是否自动ack
//            GetResponse response = channel.basicGet(queueName,false);

            long start=System.currentTimeMillis();

            while (true){
                //拉取消息
                GetResponse response = channel.basicGet(queueName,false);
                if(response == null){
                    //间隔时间，如果超过10s还没有消费到新到消息，则将消息入库，保证实效性
//                    long interval=System.currentTimeMillis()-start;
//                    if(list != null && interval>=10000){
                        //批量确认消息
//                        channel.basicAck(mqMessage.getTag(),true);
//                        list.clear();
                        //模仿业务处理
//                        Thread.sleep(1000);
//                        start=System.currentTimeMillis();
//                    }
                    break;

                }

                String str=new String(response.getBody());
//                System.out.println(str);
                ReceiveMsg receiveMsg =new ReceiveMsg();
                receiveMsg.setMessage(str);
                // 筛选消息
                SendMsg sendMsg = JSON.parseObject(str, SendMsg.class);
                if(sendMsg != null && cid.equals(sendMsg.getReceiver())){
                    receiveMsg.setTag(response.getEnvelope().getDeliveryTag());
                    list.add(receiveMsg);
//                    System.out.println(receiveMsg);
                }




//                System.out.println(mqMessage);

                //100条消息批量入库一次
//                if(list.size()%100==0){
                    // 批量手动ack  (这里发现无论是填true还是false都会确认) 所以如果不想ack则可以直接不写
//                    channel.basicAck(mqMessage.getTag(),true);
//                    list.clear();
                    //模仿业务处理
//                    Thread.sleep(1000);
//                    start=System.currentTimeMillis();
//                }

            }
        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("拉取消息异常");
        } catch (TimeoutException e) {
            e.printStackTrace();
        }finally {
            // 关闭通道
            if (channel != null && channel.isOpen()){
                try {
                    channel.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }

            // 关闭连接
            if (connection != null && connection.isOpen()){
                try {
                    connection.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }

        return list;
    }


    @GetMapping("/ack/{cid}")
    public List<ReceiveMsg> pullAck(@PathVariable String cid) {
        //创建连接工程
        ConnectionFactory connectionFactory  = new ConnectionFactory();

        connectionFactory.setHost("119.29.67.157");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("xiaohong");
        connectionFactory.setPassword("xiaohong");
        connectionFactory.setVirtualHost("my_vhost");


        Connection connection = null;
        Channel channel = null;
        //存放消息实体List
        List<ReceiveMsg> list = new ArrayList<>();
        try {
            //创建连接
            connection = connectionFactory.newConnection();
            // 通过连接获取通道Channel
            channel = connection.createChannel();
            // 通过通道创建交换机，声明队列，交换关系，路由key，发送消息和接受消息
            // 队列名称
            String queueName = "chat";
            // 手动拉取一条消息
            // 参数2 是否自动ack
//            GetResponse response = channel.basicGet(queueName,false);

            long start=System.currentTimeMillis();
            ReceiveMsg receiveMsg =new ReceiveMsg();

            while (true){
                //拉取消息
                GetResponse response = channel.basicGet(queueName, true);
                if(response == null){
                    //间隔时间，如果超过10s还没有消费到新到消息，则将消息入库，保证实效性
//                    long interval=System.currentTimeMillis()-start;
//                    if(list != null && interval>=10000){
                        //批量确认消息
//                        channel.basicAck(mqMessage.getTag(),true);
//                        list.clear();
                        //模仿业务处理
//                        Thread.sleep(1000);
//                        start=System.currentTimeMillis();
//                        return list;
                    break;
                }

                String str=new String(response.getBody());
                receiveMsg.setMessage(str);
                // 筛选消息
                SendMsg sendMsg = JSON.parseObject(str, SendMsg.class);
                if(sendMsg != null && cid.equals(sendMsg.getReceiver())){
                    receiveMsg.setTag(response.getEnvelope().getDeliveryTag());
                    list.add(receiveMsg);
//                    System.out.println(receiveMsg);
                }


                //手动ack
//                channel.basicAck(receiveMsg.getTag(),true);

                //100条消息批量入库一次
//                if(list.size()%100==0){
                    // 批量手动ack  (这里发现无论是填true还是false都会确认) 所以如果不想ack则可以直接不写
//                    channel.basicAck(mqMessage.getTag(),true);
//                    list.clear();
                    //模仿业务处理
//                    Thread.sleep(1000);
//                    start=System.currentTimeMillis();
//                }

            }

        } catch (IOException e) {
            e.printStackTrace();
            System.out.println("拉取消息异常");
        } catch (TimeoutException e) {
            e.printStackTrace();
        }finally {
            // 关闭通道
            if (channel != null && channel.isOpen()){
                try {
                    channel.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }

            // 关闭连接
            if (connection != null && connection.isOpen()){
                try {
                    connection.close();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }

        return list;
    }


}
